
/**
* 上海中赢金融信息服务有限公司
* Copyright (c) 2017, chinazyjr All Rights Reserved.
*/
package com.sys.kafka.core.consumer;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import com.sys.kafka.config.KafkaConfig;
import com.sys.kafka.config.KafkaConfig4Consumer;
import com.sys.kafka.config.KafkaConfigHolder;
import com.sys.kafka.config.KafkaConfigUtil;
import com.sys.kafka.core.consumer.listener.ConsumerExecute;
import com.sys.kafka.core.consumer.listener.MessageContainer;
import com.sys.kafka.core.consumer.listener.MessageListener;

/**
 * <b>Description：</b> TODO <br/>
 * <b>ClassName：</b> ConsumerHandler <br/>
 * <b>@author：</b> mobing <br/>
 * <b>@date：</b> 2017年7月13日 下午4:34:36 <br/>
 * <b>@version: </b> <br/>
 */
public class ConsumerHandler implements IConsumerHandler{

    private  KafkaConsumer<String, String> consumer;

    public ConsumerHandler(){}
    
    private final Properties props = new Properties();
    public void init(){
        KafkaConfigUtil.initConfig();
        KafkaConfig kafkaConfig= KafkaConfigHolder.get();
        //读取配置文件
        KafkaConfig4Consumer kafkaConfig4Consumer= kafkaConfig.getConfig4consumer();
        props.put("bootstrap.servers", kafkaConfig.getBootstrapServers());
        props.put("group.id", kafkaConfig4Consumer.getGroup_id());
        props.put("enable.auto.commit", kafkaConfig4Consumer.isEnable_auto_commit());
        props.put("auto.offset.reset",kafkaConfig4Consumer.getAuto_offset_reset());
        props.put("session.timeout.ms", kafkaConfig4Consumer.getSession_timeout_ms());
        props.put("key.deserializer", kafkaConfig4Consumer.getKey_serializer());
        props.put("value.deserializer", kafkaConfig4Consumer.getValue_serializer());
        consumer = new KafkaConsumer<String, String>(props);
    }
  

    public void subscribe(List<String> topics) {
        consumer.subscribe(topics);
    }
    
    public void subscribe(String topic) {
        List<String> topics = new ArrayList<>();
        topics.add(topic);
        consumer.subscribe(topics);
    }
    
    public void  unsubscribe(){
        consumer.unsubscribe();
    }
    
   
    public void start(){
        init();
        ConsumerExecute execute = new ConsumerExecute(consumer);
        execute.execute(5);
        
    }
    public void shutdown() {
        if (consumer != null) {
            consumer.close();
        }
       
    }

    @Override
    public void register(String topic,MessageListener listener) {
//        ConsumerExecute execute = new ConsumerExecute(consumer);
//        execute.execute(5);
        
        MessageContainer.put(topic, listener);
    }
}
